package com.innodealing.kafka.config;

import com.innodealing.kafka.processor.HermesKafkaListenerProcessor;
import com.innodealing.kafka.util.DynamicKafkaListenerRegistry;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * kafka自动配置
 * @author Yinpenghai
 */
@Configuration
@EnableKafka
@Import(HermesKafkaListenerProcessor.class)
public class KafkaAutoConfiguration {

    @Value("${hermes.kafka.bootstrap.servers}")
    private String bootstrapServers;

    @Value("${hermes.kafka.max.poll.records:20}")
    private String maxPollRecords;

    @Value("${hermes.kafka.auto.offset.reset:latest}")
    private String autoOffsetReset;

    @Value("${hermes.kafka.heartbeat.interval.ms:9000}")
    private String heartbeatIntervalMs;

    @Value("${hermes.kafka.session.timeout.ms:30000}")
    private String sessionTimeoutMs;

    @Value("${hermes.kafka.max.poll.interval.ms:300000}")
    private String maxPollIntervalMs;

    /**
     * 配置kafka消费者工厂
     * @return 消费工厂
     */
    @Bean
    public ConsumerFactory<Object, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, heartbeatIntervalMs);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollIntervalMs);
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "com.innodealing.kafka.config.CustomHashModAssignor");
        return new DefaultKafkaConsumerFactory<>(props);
    }

    /**
     * 动态监听器注册中心
     * @param consumerFactory 消费者工厂
     * @return 动态监听器注册中心
     */
    @Bean
    public DynamicKafkaListenerRegistry dynamicKafkaListenerRegistry(ConsumerFactory<Object, Object> consumerFactory) {
        return new DynamicKafkaListenerRegistry(consumerFactory);
    }
}
